﻿#include "net/mqtt/mqtt_no_ts_server.h"
# if acl
void MqttNoTsServer::init() {


    server = new MQTT_NS::server<>(
            boost::asio::ip::tcp::endpoint(
                    boost::asio::ip::tcp::v4(),
                    boost::lexical_cast<std::uint16_t>("1883")
            ),
            ioc
    );

    // 错误处理
    server->set_error_handler(
            [](MQTT_NS::error_code ec)
            {
                LOG_INFO << "error: " << ec.message();
            }
    );


    // 接收连接的处理
    server->set_accept_handler(
            [this](con_sp_t spep)
            {
                auto &ep = *spep;
                std::weak_ptr<con_t> wp(spep);

                using packet_id_t = typename std::remove_reference_t<decltype(ep)>::packet_id_t;
                LOG_INFO << "accept";

                // Pass spep to keep lifetime.
                // It makes sure wp.lock() never return nullptr in the handlers below
                // including close_handler and error_handler.
                ep.start_session(std::move(spep));

                // 正常关闭
                ep.set_close_handler([this, wp]()
                                     {
                                         LOG_INFO << "[server] closed.";
                                         auto sp = wp.lock();
                                         BOOST_ASSERT(sp);
                                         close_proc(connections, subs, sp);
                                     });

                // 超时关闭 或者 对方强制关闭 处理
                ep.set_error_handler([this, wp](MQTT_NS::error_code ec)
                                     {
                                         LOG_INFO << "[server] error: " << ec.message();
                                         auto sp = wp.lock();
                                         BOOST_ASSERT(sp);
                                         close_proc(connections, subs, sp);
                                     });

                // 权限校验
                // set MQTT level handlers
                ep.set_connect_handler(
                        [this, wp](MQTT_NS::buffer client_id, MQTT_NS::optional<MQTT_NS::buffer> username,
                                           MQTT_NS::optional<MQTT_NS::buffer> password,
                                           MQTT_NS::optional<MQTT_NS::will>,
                                           bool clean_session,
                                           std::uint16_t keep_alive)
                        {
                            using namespace MQTT_NS::literals;
                            LOG_INFO << "[server] client_id    : " << client_id;
                            LOG_INFO << "[server] username     : " << (username ? username.value() : "none"_mb);
                            LOG_INFO << "[server] password     : " << (password ? password.value() : "none"_mb);
                            LOG_INFO << "[server] clean_session: " << std::boolalpha << clean_session;
                            LOG_INFO << "[server] keep_alive   : " << keep_alive;
                            auto sp = wp.lock();
                            BOOST_ASSERT(sp);
                            connections.insert(sp);
                            sp->connack(false, MQTT_NS::connect_return_code::accepted);
                            return true;
                        }
                );

                // 关闭连接
                ep.set_disconnect_handler([this, wp]()
                                          {
                                              LOG_INFO << "[server] disconnect received.";
                                              auto sp = wp.lock();
                                              BOOST_ASSERT(sp);
                                              close_proc(connections, subs, sp);
                                          });

                //控制报文之发布消息（PUBLISH）
                // 发布确认（PUBACK）
                // 发布收到（PUBREC）
                // 发布释放（PUBREL）
                // 发布完成（PUBCOMP）

                // Qos 1级响应报文
                ep.set_puback_handler([this](packet_id_t packet_id)
                                      {
                                          LOG_INFO << "[server] puback received. packet_id: " << packet_id;
                                          return true;
                                      });

                // Qos 2级响应报文
                ep.set_pubrec_handler([this](packet_id_t packet_id)
                                      {
                                          LOG_INFO << "[server] pubrec received. packet_id: " << packet_id;
                                          return true;
                                      });

                // pu brel报文是对pu brec报文的响应
                ep.set_pubrel_handler([this](packet_id_t packet_id)
                                      {
                                          LOG_INFO << "[server] pubrel received. packet_id: " << packet_id;
                                          return true;
                                      });

                ep.set_pubcomp_handler(
                        [this]
                                (packet_id_t packet_id)
                        {
                            LOG_INFO << "[server] pubcomp received. packet_id: " << packet_id;
                            return true;
                        });

                // 收到消息的处理
                ep.set_publish_handler(
                        [this](MQTT_NS::optional<packet_id_t> packet_id,
                                MQTT_NS::publish_options pubopts,
                                MQTT_NS::buffer topic_name,
                                MQTT_NS::buffer contents)
                        {
                            LOG_INFO << "[server] publish received."
                                     << " dup: " << pubopts.get_dup()
                                     << " qos: " << pubopts.get_qos()
                                     << " retain: " << pubopts.get_retain()
                                     << " topic_name: " << topic_name
                                     << " contents: " << contents
                                     << "packet_id:" << (packet_id ? *packet_id:0);

                            // 获取tag_topic所有的索引
                            auto const &idx = subs.get<tag_topic>();

                            // 获取有相同topic的集合，遍历订阅 发送消息
                            auto r = idx.equal_range(topic_name);
                            for (; r.first != r.second; ++r.first)
                            {
                                r.first->con->publish(
                                        topic_name,
                                        contents,
                                        std::min(r.first->qos_value, pubopts.get_qos())
                                );
                            }
                            return true;
                        });

                // 订阅
                ep.set_subscribe_handler(
                        [this, wp](packet_id_t packet_id, std::vector<MQTT_NS::subscribe_entry> entries)
                        {
                            LOG_INFO << "[server] subscribe received. packet_id: " << packet_id;
                            std::vector<MQTT_NS::suback_return_code> res;
                            res.reserve(entries.size());
                            auto sp = wp.lock();
                            BOOST_ASSERT(sp);
                            for (auto const &e: entries)
                            {
                                LOG_INFO << "[server] topic_filter: " << e.topic_filter
                                         << " qos: " << e.subopts.get_qos();
                                res.emplace_back(MQTT_NS::qos_to_suback_return_code(e.subopts.get_qos()));
                                subs.emplace(std::move(e.topic_filter), sp, e.subopts.get_qos());
                            }
                            sp->suback(packet_id, res);
                            return true;
                        }
                );

                // 取消订阅
                ep.set_unsubscribe_handler(
                        [this, wp](packet_id_t packet_id,
                                    std::vector<MQTT_NS::unsubscribe_entry> entries)
                        {
                            LOG_INFO << "[server] unsubscribe received. packet_id: " << packet_id;
                            auto sp = wp.lock();
                            for (auto const &e: entries)
                            {
                                auto it = subs.find(std::make_tuple(sp, e.topic_filter));
                                if (it != subs.end())
                                {
                                    subs.erase(it);
                                }
                            }
                            BOOST_ASSERT(sp);
                            sp->unsuback(packet_id);
                            return true;
                        }
                );
            }
    );

}

void MqttNoTsServer::run() {
    server->listen();

    ioc.run();
}

MqttNoTsServer::~MqttNoTsServer() {

    if(server){
        delete server;
    }
}
# endif